package com.itcast.spark

import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
 * DESC:当前的案例主要是接受的Socket的数据，需要指明主机名和端口号接受数据
 * 注意在这里SparkStreaming采取的方式是receiver进程拉取数据源的数据，进行不同阶段的处理
 */
object SparkStremingTCP {
  def main(args: Array[String]): Unit = {
    //1-环境准备
    val conf: SparkConf = new SparkConf().setAppName("SparkStremingTCP").setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    //这里就是指定配置项将数据按照5秒为周期进行处理
    val ssc = new StreamingContext(sc, Seconds(5))
    //2-准备socket
    val receiveData: ReceiverInputDStream[String] = ssc.socketTextStream("node01", 9999, StorageLevel.MEMORY_AND_DISK_2)
    //3-接下来对该数据进行简单的wordcount的统计
    val flatMapDS: DStream[String] = receiveData.flatMap(_.split(" "))
    val mapDS: DStream[(String, Int)] = flatMapDS.map(x => (x, 1))
    val reduceDS: DStream[(String, Int)] = mapDS.reduceByKey((x: Int, y: Int) => x + y)
    //4-如果想对当前的结果进行排序的化，请问如何实现-transform(func)
    //5-执行结果的输出
    reduceDS.print()
    //6-开启程序执行---Start the execution of the streams.开启streams的执行
    ssc.start()
    //7-等到程序停止---等待执行器的停止，用户或程序触发任何异常都会引起当前程序停止
    ssc.awaitTermination()
    //8-停止streamingcontext
    //第一个参数就是停止saprkcontext
    //第二个参数就是优雅停止，等到所有的接收器接受的数据处理完毕之后再停止
    ssc.stop(true,true)


  }
}
